Elasticsearch （二）

logstash

logstash基本语法组成

logstash主要由三部分组成：input、filter、output。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
//输入插件
input{

    //文件类型
    file{
	#path属性接受的参数是一个数组，其含义是标明需要读取的文件位置
	path => [‘pathA’，‘pathB’]

	#表示多就去path路径下查看是够有新的文件产生。默认是15秒检查一次。
	discover_interval => 15

	#排除那些文件，也就是不去读取那些文件
	exclude => [‘fileName1’,‘fileNmae2’]

	#被监听的文件多久没更新后断开连接不在监听，默认是一个小时。
	close_older => 3600

	#在每次检查文件列 表的时候， 如果一个文件的最后 修改时间 超过这个值， 就忽略这个文件。 默认一天。
	ignore_older => 86400

	#logstash 每隔多 久检查一次被监听文件状态（ 是否有更新） ， 默认是 1 秒。
	stat_interval => 1

	#sincedb记录数据上一次的读取位置的一个index
	sincedb_path => ’$HOME/. sincedb‘

	#logstash 从什么 位置开始读取文件数据， 默认是结束位置 也可以设置为：beginning 从头开始
	start_position => ‘beginning’

	#注意：这里需要提醒大家的是，如果你需要每次都从同开始读取文件的话，关设置start_position => beginning是没有用的，你可以选择sincedb_path 定义为 /dev/null
    }  
    
    //数据库类型
    jdbc{
	#jdbc sql server 驱动,各个数据库都有对应的驱动，需自己下载
	jdbc_driver_library => "/etc/logstash/driver.d/sqljdbc_2.0/enu/sqljdbc4.jar"

	#jdbc class 不同数据库有不同的 class 配置
	jdbc_driver_class => "com.microsoft.sqlserver.jdbc.SQLServerDriver"

	#配置数据库连接 ip 和端口，以及数据库    
	jdbc_connection_string => "jdbc:sqlserver://200.200.0.18:1433;databaseName=test_db"

	#配置数据库用户名
	jdbc_user =>   

	#配置数据库密码
	jdbc_password =>

	#上面这些都不重要，要是这些都看不懂的话，你的老板估计要考虑换人了。重要的是接下来的内容。
	# 定时器 多久执行一次SQL，默认是一分钟
	# schedule => 分 时 天 月 年  
	# schedule => * 22  *  *  * 表示每天22点执行一次
	schedule => "* * * * *"

	#是否清除 last_run_metadata_path 的记录,如果为真那么每次都相当于从头开始查询所有的数据库记录
	clean_run => false

	#是否需要记录某个column 的值,如果 record_last_run 为真,可以自定义我们需要表的字段名称，
	#此时该参数就要为 true. 否则默认 track 的是 timestamp 的值.
	use_column_value => true

	#如果 use_column_value 为真,需配置此参数. 这个参数就是数据库给出的一个字段名称。当然该字段必须是递增的，可以是 数据库的数据时间这类的
	tracking_column => create_time

	#是否记录上次执行结果, 如果为真,将会把上次执行到的 tracking_column 字段的值记录下来,保存到 last_run_metadata_path 指定的文件中
	record_last_run => true

	#们只需要在 SQL 语句中 WHERE MY_ID > :last_sql_value 即可. 其中 :last_sql_value 取得就是该文件中的值
	last_run_metadata_path => "/etc/logstash/run_metadata.d/my_info"

	#是否将字段名称转小写。
	#这里有个小的提示，如果你这前就处理过一次数据，并且在Kibana中有对应的搜索需求的话，还是改为true，
	#因为默认是true，并且Kibana是大小写区分的。准确的说应该是ES大小写区分
	lowercase_column_names => false

	#你的SQL的位置，当然，你的SQL也可以直接写在这里。
	#statement => SELECT * FROM tabeName t WHERE  t.creat_time > :last_sql_value
	statement_filepath => "/etc/logstash/statement_file.d/my_info.sql"

	#数据类型，标明你属于那一方势力。单了ES哪里好给你安排不同的山头。
	type => "my_info"
    }
    #注意：外载的SQL文件就是一个文本文件就可以了，还有需要注意的是，一个jdbc{}插件就只能处理一个SQL语句，
    #如果你有多个SQL需要处理的话，只能在重新建立一个jdbc{}插件。

    //redis
    redis {
	codec =>             #可选项，默认是plain，可设置其他编码方式。

	discover_interval => #可选项，logstash多久检查一下path下有新文件，默认15s。

	exclude =>           #可选项，排除path下不想监听的文件。

	sincedb_path =>      #可选项，记录文件以及文件读取信息位置的数据文件。~/.sincedb_xxxx

	sincedb_write_interval => #可选项，logstash多久写一次sincedb文件，默认15s.

	stat_interval =>          #可选项，logstash多久检查一次被监听文件的变化，默认1s。

	start_position =>         #可选项，logstash从哪个位置读取文件数据，默认从尾部，值为：end。初次导入，设置为：beginning。

	path =>                   #必选项，配置文件路径，可定义多个。

	tags =>                   #可选项，在数据处理过程中，由具体的插件来添加或者删除的标记。

	type =>                   #可选项，自定义处理时间类型。比如nginxlog。
    }
}

filter{
	过滤器插件
	过滤器提供的并不单单是过滤的功能，还可以对进入过滤器的原始数据进行复杂的逻辑处理，甚至添加独特的新事件到后续流程中
}

outer{

　　输出插件
    elasticsearch {
        id => "log"
        hosts => ["elasticsearch:9200"]
        index => "logstash-%{type}-%{+YYYY.MM.dd}"
    }
}